package com.hb.resource.producer;

import com.google.common.collect.Lists;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.ArrayList;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.IntStream;

/**
 * @author
 * @date:2023/4/25 15:20
 */
public class KafkaCustomProducer {

    public static void main(String[] args) {
        // 初始化配置
        Properties properties = initKafka();
//		KafkaProducer

        // 创建生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        String topic = "testTopicNew2";

        IntStream.rangeClosed(1, 10).forEach(i->{
            String message = "Hello world_" + i;
            normalSend(kafkaProducer, topic, message);
//			producerSendWithCallback(kafkaProducer, topic, message);
//			producerSendWithCallbackWithPartition(kafkaProducer, topic, message);

            System.out.println(message);
        });
        kafkaProducer.flush();
        kafkaProducer.close();// 会调用拦截器中的close方法

    }

    /**
     * 初始化卡夫卡
     *
     * @return {@link Properties}
     */
    public static Properties initKafka() {
        Properties properties = new Properties();
        // dev环境
//		properties.put("bootstrap.servers", "10.18.0.73:31090,10.18.0.73:31091,10.18.0.73:31092");
        // 本地
        properties.put("bootstrap.servers", "127.0.0.1:9092");
        properties.put("acks", "all");
        properties.put("retries", 3);
        properties.put("batch.size", 16384);
//		等待时间, 默认值60000MS, 一分钟, 坑爹哦.
//		properties.put("max.block.ms", 100);
        properties.put("buffer.memory", 33554432);
        properties.put("linger.ms", 1);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 自定义分区实现类.
        properties.put("partitioner.class", "com.hb.resource.config.MyPartitioner");

        ArrayList<String> incerceptors = Lists.newArrayList("com.hb.resource.interceptor.TmeInterceptor", "com.hb.resource.interceptor.CounterInterceptor");
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, incerceptors);
        return properties;
    }

    /**
     * 正常发送
     *
     * @param producer 生产商
     * @param topic    主题
     * @param message  消息
     */
    public static void normalSend(KafkaProducer<String, String> producer, String topic, String message) {
        Future<RecordMetadata> future = producer.send(new ProducerRecord<>(topic, message));
    }

    /**
     * 生产者发送与回调
     *
     * @param producer 生产商
     * @param topic    主题
     * @param message  消息
     */
    public static void producerSendWithCallback(KafkaProducer<String, String> producer, String topic, String message) {
        producer.send(new ProducerRecord<>(topic, message), (metadata, exception) -> {
            if (Objects.isNull(exception)) {
                System.out.println(metadata.partition() + "--" + metadata.offset());
                System.out.println(metadata.toString());
            } else {
                exception.printStackTrace();
            }
        });
    }

    /**
     * 生产者发送与分区进行回调
     *
     * @param producer 生产商
     * @param topic    主题
     * @param message  消息
     */
    public static void producerSendWithCallbackWithPartition(KafkaProducer<String, String> producer, String topic, String message) {
        producer.send(new ProducerRecord<>(topic, 0, "hello", message), (metadata, exception) -> {
            if (Objects.isNull(exception)) {
                System.out.println(metadata.partition() + "--"
                        + metadata.offset());
                System.out.println(metadata.toString());
            } else {
                exception.printStackTrace();
            }
        });
    }


    /**
     * 正常发送diy分区
     *
     * @param producer 生产商
     * @param topic    主题
     * @param message  消息
     */
    public static void normalSendDiyPartition(KafkaProducer<String, String> producer, String topic, String message) {
        producer.send(new ProducerRecord<>(topic, message));
    }

    /**
     * 同步发送, 用于保证全局有序.
     *
     * @param producer 生产商
     * @param topic    主题
     * @param message  消息
     * @throws ExecutionException   执行异常
     * @throws InterruptedException 中断异常
     */
    public static void syncSend(KafkaProducer<String, String> producer, String topic, String message) throws ExecutionException, InterruptedException {
        Future<RecordMetadata> future = producer.send(new ProducerRecord<>(topic, message));
        future.get();
    }
}
